package com.zfei.gfb.device;

import com.google.common.collect.ImmutableSet;
import com.zfei.gfb.api.domain.EventTuple;
import com.zfei.gfb.api.domain.InboxTuple;
import com.zfei.gfb.api.domain.KeyTuple;
import com.zfei.gfb.api.domain.OrderType;
import com.zfei.gfb.util.Strings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.DefaultTuple;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
import org.springframework.data.redis.connection.StringRedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.*;

@Component
public class InboxCache {

    public static final EventTuple COMPLETE = new EventTuple("COMPLETE", -1L);
    public static final EventTuple MAXLENGTH = new EventTuple("MAXLENGTH", -2L);
    public static final ImmutableSet<EventTuple> FLAGS = ImmutableSet.of(COMPLETE, MAXLENGTH);
    private static final Logger logger = LoggerFactory.getLogger(InboxCache.class);
    private StringRedisTemplate redis;

    public InboxCache(RedisConnectionFactory factory) {
        this.redis = new StringRedisTemplate(factory);
    }

    public StringRedisTemplate getRedis() {
        return redis;
    }

    public List<KeyTuple> convert(String name, Collection<TypedTuple<String>> typedTuples) {
        List<KeyTuple> result = new ArrayList<>();
        if (typedTuples != null && typedTuples.size() > 0) {
            for (TypedTuple<String> typedTuple : typedTuples) {
                KeyTuple inboxTuple = new KeyTuple();
                inboxTuple.setName(name);
                inboxTuple.setEventId(typedTuple.getValue());
                inboxTuple.setSearchNum(Objects.requireNonNull(typedTuple.getScore()).doubleValue());
                result.add(inboxTuple);
            }
        }
        return result;
    }

    public Set<Tuple> convert(Collection<InboxTuple> inboxTuples) {
        Set<Tuple> result = new LinkedHashSet<>();
        for (InboxTuple inboxTuple : inboxTuples) {
            result.add(new DefaultTuple(Strings.serialize(inboxTuple.getEventId()), inboxTuple.getEventTime().doubleValue()));
        }
        return result;
    }

    public List<KeyTuple> list(String name, long minTime, long maxTime, int offset, int count, int orderType) {
        long min = minTime + 1;
        long max = maxTime - 1;
        Set<TypedTuple<String>> typedTuples;
        if (orderType == OrderType.DESC) {
            typedTuples = this.getRedis().opsForZSet().reverseRangeByScoreWithScores(name, min, max, offset, count);
        } else {
            typedTuples = this.getRedis().opsForZSet().rangeByScoreWithScores(name, min, max, offset, count);
        }

        return convert(name, typedTuples);
    }

    public List<KeyTuple> list(String name, long start, long stop, int orderType) {
        Set<TypedTuple<String>> typedTuples;
        if (orderType == OrderType.DESC) {
            typedTuples = this.getRedis().opsForZSet().reverseRangeWithScores(name, start, stop);
        } else {
            typedTuples = this.getRedis().opsForZSet().rangeWithScores(name, start, stop);
        }

        return convert(name, typedTuples);
    }

    public long count(String name, long minTime, long maxTime) {
        long min = minTime + 1;
        long max = maxTime - 1;
        Long count = this.getRedis().opsForZSet().count(name, min, max);
        return count != null ? count : 0;
    }

    public KeyTuple get(String name, String value) {
        Double score = this.getRedis().opsForZSet().score(name, value);
        if (score != null) {
            KeyTuple inboxTuple = new KeyTuple();
            inboxTuple.setName(name);
            inboxTuple.setEventId(value);
            inboxTuple.setSearchNum(score.doubleValue());
            return inboxTuple;
        }
        return null;
    }

    public List<InboxTuple> get(String name, List<String> values) {
        List<InboxTuple> result = new ArrayList<>();
        List<Object> scores = this.redis.executePipelined((RedisCallback<Object>) connection -> {
            StringRedisConnection conn = (StringRedisConnection) connection;
            for (String value : values) {
                conn.zScore(name, value);
            }
            return null;
        });
        if (scores != null && scores.size() == values.size()) {
            for (int i = 0; i < values.size(); i++) {
                Double score = (Double) scores.get(i);
                if (score != null) {
                    InboxTuple inboxTuple = new InboxTuple();
                    inboxTuple.setName(name);
                    inboxTuple.setEventId(values.get(i));
                    inboxTuple.setEventTime(score.longValue());
                    result.add(inboxTuple);
                }
            }
        }
        return result;
    }

    public KeyTuple head(String name) {
        List<KeyTuple> tuples = this.list(name, 0, 0, OrderType.ASC);
        if (tuples != null && tuples.size() > 0) {
            return tuples.get(0);
        }
        return null;
    }

    public KeyTuple tail(String name) {
        List<KeyTuple> tuples = this.list(name, 0, 0, OrderType.DESC);
        if (tuples != null && tuples.size() > 0) {
            return tuples.get(0);
        }
        return null;
    }

    public boolean exists(String name) {
        KeyTuple head = this.head(name);
        return head != null;
    }


    public boolean clear(String name) {
        Boolean ret = this.redis.delete(name);
        return ret != null && ret;
    }

    public void insert(KeyTuple inboxTuple, Duration timeout) {
        this.redis.executePipelined((RedisCallback<Object>) connection -> {
            StringRedisConnection conn = (StringRedisConnection) connection;
            conn.zAdd(inboxTuple.getName(), inboxTuple.getSearchNum(), inboxTuple.getEventId());
            conn.expire(inboxTuple.getName(), timeout.getSeconds());
            return null;
        });
    }

    public void insert(List<InboxTuple> inboxTuples, Duration timeout) {
        if (inboxTuples == null || inboxTuples.isEmpty()) {
            return;
        }

        Map<String, List<InboxTuple>> map = new HashMap<>();
        for (InboxTuple inboxTuple : inboxTuples) {
            List<InboxTuple> value = map.computeIfAbsent(inboxTuple.getName(), s -> new ArrayList<>());
            value.add(inboxTuple);
        }

        for (Map.Entry<String, List<InboxTuple>> entry : map.entrySet()) {
            String name = entry.getKey();
            Set<Tuple> tuples = convert(entry.getValue());
            if (tuples != null && tuples.size() > 0) {
                this.redis.executePipelined((RedisCallback<Object>) connection -> {
                    byte[] rawKey = Strings.serialize(name);
                    connection.zAdd(rawKey, tuples);
                    connection.expire(rawKey, timeout.getSeconds());
                    return null;
                });
            }
        }
    }

    public void insert(InboxTuple inboxTuple) {
        Long expire = this.redis.getExpire(inboxTuple.getName());
        if (expire == null || expire <= 0L) {
            this.redis.executePipelined((RedisCallback<Object>) connection -> {
                StringRedisConnection conn = (StringRedisConnection) connection;
                conn.zAdd(inboxTuple.getName(), inboxTuple.getEventTime(), inboxTuple.getEventId());
                conn.expire(inboxTuple.getName(), Duration.ofHours(72).getSeconds());
                return null;
            });
        } else {
            this.redis.opsForZSet().add(inboxTuple.getName(), inboxTuple.getEventId(), inboxTuple.getEventTime());
        }
    }

    public void insert(InboxTuple inboxTuple, Duration timeout) {
        this.redis.executePipelined((RedisCallback<Object>) connection -> {
            StringRedisConnection conn = (StringRedisConnection) connection;
            conn.zAdd(inboxTuple.getName(), inboxTuple.getEventTime(), inboxTuple.getEventId());
            conn.expire(inboxTuple.getName(), timeout.getSeconds());
            return null;
        });
    }

    public void insert(List<InboxTuple> inboxTuples) {
        if (inboxTuples == null || inboxTuples.isEmpty()) {
            return;
        }

        for (InboxTuple inboxTuple : inboxTuples) {
            insert(inboxTuple);
        }

    }

    public void delete(String name, String value) {
        if (value == null) {
            return;
        }

        this.redis.opsForZSet().remove(name, value);
    }

    public void delete(String key, List<String> values) {
        if (values == null || values.isEmpty()) {
            return;
        }

        Object[] del = new Object[values.size()];
        for (int i = 0; i < values.size(); i++) {
            del[i] = values.get(i);
        }
        this.redis.opsForZSet().remove(key, del);
    }

    public void delete(InboxTuple inboxTuple) {
        if (inboxTuple == null) {
            return;
        }

        this.delete(inboxTuple.getName(), inboxTuple.getEventId());
    }

    public void delete(List<InboxTuple> inboxTuples) {
        if (inboxTuples == null || inboxTuples.isEmpty()) {
            return;
        }

        Map<String, List<String>> map = new HashMap<>();
        for (InboxTuple inboxTuple : inboxTuples) {
            List<String> value = map.computeIfAbsent(inboxTuple.getName(), s -> new ArrayList<>());
            value.add(inboxTuple.getEventId());
        }

        for (Map.Entry<String, List<String>> entry : map.entrySet()) {
            this.delete(entry.getKey(), entry.getValue());
        }
    }

    /**
     * 线上redis版本不支持unionAndStore命令
     */
    @Deprecated
    public void merge(String key, List<String> mergeBys) {
        if (mergeBys == null || mergeBys.isEmpty()) {
            return;
        }
        this.getRedis().opsForZSet().unionAndStore(null, mergeBys, key);
    }
}
